package com.poetic.network.flow;

import com.poetic.network.flow.domain.UserBehavior;
import com.poetic.network.flow.enums.BehaviorEnum;
import com.poetic.network.flow.process.UVCountWithBloom;
import com.poetic.network.flow.trigger.MyTrigger;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * <pre>
 *
 *  们不需要完整地存储用户 ID 的信息，只要知道他在
 * 不在就行了。所以其实我们可以进行压缩处理，用一位（bit）就可以表示一个用户
 * 的状态。这个思想的具体实现就是布隆过滤器（Bloom Filter）。
 * 本质上布隆过滤器是一种数据结构，比较巧妙的概率型数据结构（probabilistic
 * data structure），特点是高效地插入和查询，可以用来告诉你 “某样东西一定不存
 * 在或者可能存在”。
 * 它本身是一个很长的二进制向量，既然是二进制的向量，那么显而易见的，存
 * 放的不是 0，就是 1。相比于传统的 List、Set、Map 等数据结构，它更高效、占用
 * 空间更少，但是缺点是其返回的结果是概率性的，而不是确切的
 *
 *
 * Created by lianghuikun on 2020-09-15.
 * </pre>
 *
 * @author lianghuikun
 */
public class UvWithBloomFilterTask {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设定 Time 类型为 EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // 为了打印到控制台的结果不乱序，我们配置全局的并发为 1，这里改变并发对结果正确性没有影响
        env.setParallelism(1);
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/UserBehavior.csv";
        env.readTextFile(path)
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] data = value.split("\\,");
                        return UserBehavior.builder()
                                .userId(Long.valueOf(data[0]))
                                .itemId(Long.valueOf(data[1]))
                                .categoryId(Integer.valueOf(data[2]))
                                .behavior(data[3])
                                .timestamp(Long.valueOf(data[4]))
                                .build();
                    }
                })
                .filter(new FilterFunction<UserBehavior>() {
                    @Override
                    public boolean filter(UserBehavior value) throws Exception {
                        return value.getBehavior().equals(BehaviorEnum.pv.name());
                    }
                })
                .map(new MapFunction<UserBehavior, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(UserBehavior value) throws Exception {
                        return Tuple2.of("dummyKey", value.getUserId());
                    }
                })
                .keyBy(new KeySelector<Tuple2<String, Long>, Long>() {
                    @Override
                    public Long getKey(Tuple2<String, Long> value) throws Exception {
                        return value.f1;
                    }
                })
                .timeWindow(Time.hours(1))
                // 自定义窗口触发规则,来一条数据就触发一次操作，写入redis，不能等窗口所有数据都来了才触发，因为内存放不下
                .trigger(new MyTrigger())
                // 自定义窗口处理规则
                .process(new UVCountWithBloom())
                .print();

        env.execute("Unique Visitor with bloom Job");
    }
}
